package hwactor

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable


class CheckNodeManagerThread extends  Thread {

  override def run(): Unit ={
    while(true) {
      Thread.sleep(3000)

      val currentTime = System.currentTimeMillis()
      for((host, time) <- MyCollection.mapHeartBeat){
          if(currentTime - time > 10*1000*1000){
            println("Host:" + host + " Time:" + time + "下线")
            MyCollection.mapHeartBeat.remove(host)
            MyCollection.mapRegister.remove(host)
          }
          println("Host:" + host + " Time:" + time + "在线")
      }
    }
  }
}

object MyCollection{
  var mapRegister  = new mutable.HashMap[String, RegisterClass]()
  var mapHeartBeat = new mutable.HashMap[String, Long]()
}


class MyResourceManager extends  Actor {

  def handleRegisterMsg(nmid : String , cpu:Int, mem : Int): Unit ={
    println("host : " + nmid + " cpu: " + cpu + " mem:" + mem)
    // 将消息发入到集合中
    MyCollection.mapRegister.put(nmid, new RegisterClass(nmid, cpu, mem))
    sender() ! new RegisteredNodeManager("MyResourceManager")
  }

  def handleHearbeat(nmid : String, time:Long): Unit ={
    MyCollection.mapHeartBeat.put(nmid, time)
    println("heartbeat : " + nmid)
  }

  override def receive: Receive = {
    case RegisterMsg(nmid : String , cpu:Int, mem : Int) =>{
      handleRegisterMsg(nmid, cpu, mem)
    }
    case HearbeatMsg(nmid : String, time: Long ) =>{
      println("heartbeat : " + nmid + " " + time)
      handleHearbeat(nmid, time)
    }
  }
}

object MyResourceManager {
  def main(args: Array[String]): Unit = {
    val str=
      """
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname =localhost
        |akka.remote.netty.tcp.port=19999
      """.stripMargin
    val config = ConfigFactory.parseString(str)
    val actorSystem = ActorSystem("MyResourceManagerActorSystem", config)
    actorSystem.actorOf(Props(new MyResourceManager()), "MyResourceManagerActor")
    new CheckNodeManagerThread().start()
  }
}
